package com.bw.gmall.realtime.app.dws.dws_02;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.api.java.tuple.Tuple8;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

public class DwsTradeOrderRefundWindow {
    public static void main(String[] args) throws Exception {
        // 1. 获取流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2.读取DWD层成功支付主题数据创建流
        String topic = "dwd_trade_order_refund_a";
        String groupId = "dwd_trade_order_refund_a_250703";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));


//        // 3. 解析JSON为JSONObject
//        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaDS.map(new MapFunction<String, JSONObject>() {
//            @Override
//            public JSONObject map(String value) throws Exception {
//                return JSONObject.parseObject(value);
//            }
//        });
//
//        // 4. 分配事件时间和水位线（基于create_time）
//        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonStream.assignTimestampsAndWatermarks(
//                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
//                            @Override
//                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
//                                // 转换create_time为时间戳（格式：yyyy-MM-dd HH:mm:ss）
//                                return java.sql.Timestamp.valueOf(element.getString("create_time")).getTime();
//                            }
//                        })
//        );
//
//        // 5. 提取关键字段：店铺ID、退款类型编码（1501/1502）、退款金额
//        // 输出格式：Tuple4(shopId, refundTypeCode, refundAmount, 1)
//        SingleOutputStreamOperator<Tuple4<String, String, Double, Integer>> refundTupleStream = withWatermarkStream.map(
//                new MapFunction<JSONObject, Tuple4<String, String, Double, Integer>>() {
//                    @Override
//                    public Tuple4<String, String, Double, Integer> map(JSONObject value) throws Exception {
//                        String shopId = value.getString("shop_id");
//                        String refundTypeCode = value.getString("refund_type_code"); // 1501=失败，1502=成功
//                        Double refundAmount = Double.parseDouble(value.getString("refund_amount"));
//                        return new Tuple4<>(shopId, refundTypeCode, refundAmount, 1);
//                    }
//                }
//        );
//
//        // 6. 按店铺ID分组，1天窗口聚合三类金额
//        SingleOutputStreamOperator<Tuple4<String, Double, Double, Double>> resultStream = refundTupleStream
//                .keyBy(tuple -> tuple.f0) // 按店铺ID分组
//                .window(TumblingEventTimeWindows.of(Time.days(1))) // 1天滚动窗口
//                .aggregate(
//                        // 累加器：(总金额, 成功金额, 失败金额)
//                        new AggregateFunction<
//                                Tuple4<String, String, Double, Integer>,
//                                Tuple4<String, Double, Double, Double>,
//                                Tuple4<String, Double, Double, Double>>() {
//
//                            @Override
//                            public Tuple4<String, Double, Double, Double> createAccumulator() {
//                                return new Tuple4<>("", 0.0, 0.0, 0.0); // (店铺ID, 总金额, 成功金额, 失败金额)
//                            }
//
//                            @Override
//                            public Tuple4<String, Double, Double, Double> add(
//                                    Tuple4<String, String, Double, Integer> value,
//                                    Tuple4<String, Double, Double, Double> accumulator) {
//
//                                String shopId = value.f0;
//                                String typeCode = value.f1;
//                                Double amount = value.f2;
//
//                                // 累加总金额
//                                double total = accumulator.f1 + amount;
//                                // 区分成功/失败金额（基于1501=退款失败，1502=退款成功）
//                                double success = accumulator.f2;
//                                double fail = accumulator.f3;
//
//                                if ("1502".equals(typeCode)) { // 1502=成功退款
//                                    success += amount;
//                                } else if ("1501".equals(typeCode)) { // 1501=失败退款
//                                    fail += amount;
//                                }
//
//                                return new Tuple4<>(shopId, total, success, fail);
//                            }
//
//                            @Override
//                            public Tuple4<String, Double, Double, Double> getResult(Tuple4<String, Double, Double, Double> accumulator) {
//                                return accumulator;
//                            }
//
//                            @Override
//                            public Tuple4<String, Double, Double, Double> merge(
//                                    Tuple4<String, Double, Double, Double> a,
//                                    Tuple4<String, Double, Double, Double> b) {
//                                // 合并两个累加器（用于窗口合并）
//                                return new Tuple4<>(
//                                        a.f0, // 店铺ID
//                                        a.f1 + b.f1, // 总金额
//                                        a.f2 + b.f2, // 成功金额
//                                        a.f3 + b.f3  // 失败金额
//                                );
//                            }
//                        },
//                        // 窗口函数：输出最终结果
//                        new ProcessWindowFunction<
//                                Tuple4<String, Double, Double, Double>,
//                                Tuple4<String, Double, Double, Double>,
//                                String,
//                                TimeWindow>() {
//
//                            @Override
//                            public void process(
//                                    String shopId, // 分组键：店铺ID
//                                    Context context,
//                                    Iterable<Tuple4<String, Double, Double, Double>> elements,
//                                    Collector<Tuple4<String, Double, Double, Double>> out) {
//
//                                Tuple4<String, Double, Double, Double> result = elements.iterator().next();
//                                out.collect(result);
//                            }
//                        }
//                );
//
//        // 7. 输出结果（格式：(店铺ID, 总退款金额, 成功退款金额, 失败退款金额)）
//        resultStream.print("各店铺退款金额统计：");
//
//        // 5. 提取店铺ID，仅标记失败订单（1501=失败）为1，成功订单为0
//        SingleOutputStreamOperator<Tuple2<String, Integer>> shopFailCountStream = withWatermarkStream.map(
//                value -> {
//                    String shopId = value.getString("shop_id");
//                    String refundType = value.getString("refund_type_code");
//                    // 失败订单标记为1，否则为0
//                    int failFlag = "1501".equals(refundType) ? 1 : 0;
//                    return new Tuple2<>(shopId, failFlag); // (店铺ID, 失败订单标记)
//                }
//        );
//
//        // 6. 按店铺分组，1天窗口聚合失败退款个数
//        SingleOutputStreamOperator<Tuple2<String, Integer>> resultStream1 = shopFailCountStream
//                .keyBy(tuple -> tuple.f0)
//                .window(TumblingEventTimeWindows.of(Time.days(1)))
//                .aggregate(
//                        // 累加器：累计失败订单数
//                        new AggregateFunction<Tuple2<String, Integer>, Integer, Integer>() {
//                            @Override
//                            public Integer createAccumulator() {
//                                return 0;
//                            }
//
//                            @Override
//                            public Integer add(Tuple2<String, Integer> value, Integer accumulator) {
//                                return accumulator + value.f1; // 失败订单加1，成功订单加0
//                            }
//
//                            @Override
//                            public Integer getResult(Integer accumulator) {
//                                return accumulator;
//                            }
//
//                            @Override
//                            public Integer merge(Integer a, Integer b) {
//                                return a + b;
//                            }
//                        },
//                        // 窗口函数：输出(店铺ID, 退款失败个数)
//                        new ProcessWindowFunction<Integer, Tuple2<String, Integer>, String, TimeWindow>() {
//                            @Override
//                            public void process(String shopId, Context context, Iterable<Integer> elements, Collector<Tuple2<String, Integer>> out) {
//                                out.collect(new Tuple2<>(shopId, elements.iterator().next()));
//                            }
//                        }
//                );
//
//        // 7. 输出结果
//        resultStream1.print("各店铺退款失败个数：");

//
//// 3. 解析JSON并分配事件时间
//        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaDS.map(JSONObject::parseObject);
//
//        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonStream.assignTimestampsAndWatermarks(
//                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                        .withTimestampAssigner((element, recordTimestamp) ->
//                                java.sql.Timestamp.valueOf(element.getString("create_time")).getTime()
//                        )
//        );
//
//        // 4. 提取关键信息：(店铺ID, 退款类型, 退款金额, 处理中标记, 失败标记)
//        SingleOutputStreamOperator<Tuple6<String, String, Double, Integer, Integer, Integer>> refundTupleStream =
//        withWatermarkStream.map(
//                value -> {
//                    String shopId = value.getString("shop_id");
//                    String refundType = value.getString("refund_type_code");
//                    Double amount = Double.parseDouble(value.getString("refund_amount"));
//
//                    // 标记退款状态（根据业务规则调整）
//                    int isProcessing = 0; // 处理中标记
//                    int isSuccess = 0;    // 成功标记
//                    int isFail = 0;       // 失败标记
//
//                    if ("1502".equals(refundType)) {
//                        isSuccess = 1;  // 1502=成功
//                    } else if ("1501".equals(refundType)) {
//                        isFail = 1;     // 1501=失败
//                    } else {
//                        isProcessing = 1; // 其他状态视为处理中
//                    }
//
//                    return new Tuple6<>(shopId, refundType, amount, isProcessing, isSuccess, isFail);
//                }
//        );

        // 5. 窗口聚合：统计所有指标
//        SingleOutputStreamOperator<Tuple6<String, Double, Double, Double, Integer, Integer>> resultStream = refundTupleStream
//                .keyBy(tuple -> tuple.f0)
//                .window(TumblingEventTimeWindows.of(Time.days(1)))
//                .aggregate(
//                        new AggregateFunction<
//                                Tuple6<String, String, Double, Integer, Integer, Integer>,
//                                Tuple6<String, Double, Double, Double, Integer, Integer>,
//                                Tuple6<String, Double, Double, Double, Integer, Integer>>() {
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> createAccumulator() {
//                                return new Tuple6<>("", 0.0, 0.0, 0.0, 0, 0);
//                                // (店铺ID, 总金额, 成功金额, 失败金额, 处理中个数, 失败个数)
//                            }
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> add(
//                                    Tuple6<String, String, Double, Integer, Integer, Integer> value,
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> accumulator) {
//
//                                double totalAmount = accumulator.f1 + value.f2;
//                                double successAmount = accumulator.f2 + (value.f4 == 1 ? value.f2 : 0);
//                                double failAmount = accumulator.f3 + (value.f5 == 1 ? value.f2 : 0);
//                                int processingCount = accumulator.f4 + value.f3;
//                                int failCount = accumulator.f5 + value.f5;
//
//                                return new Tuple6<>(value.f0, totalAmount, successAmount, failAmount, processingCount, failCount);
//                            }
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> getResult(
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> accumulator) {
//                                return accumulator;
//                            }
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> merge(
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> a,
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> b) {
//                                return new Tuple6<>(
//                                        a.f0,
//                                        a.f1 + b.f1,
//                                        a.f2 + b.f2,
//                                        a.f3 + b.f3,
//                                        a.f4 + b.f4,
//                                        a.f5 + b.f5
//                                );
//                            }
//                        },
//                        new ProcessWindowFunction<
//                                Tuple6<String, Double, Double, Double, Integer, Integer>,
//                                Tuple6<String, Double, Double, Double, Integer, Integer>,
//                                String,
//                                TimeWindow>() {
//
//                            @Override
//                            public void process(
//                                    String shopId,
//                                    Context context,
//                                    Iterable<Tuple6<String, Double, Double, Double, Integer, Integer>> elements,
//                                    Collector<Tuple6<String, Double, Double, Double, Integer, Integer>> out) {
//
//                                out.collect(elements.iterator().next());
//                            }
//                        }
//                );



        // 原 Tuple6：(店铺ID, 总金额, 成功金额, 失败金额, 处理中个数, 失败个数)
//// 3. 解析JSON并分配事件时间
//        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaDS.map(new MapFunction<String, JSONObject>() {
//            @Override
//            public JSONObject map(String value) throws Exception {
//                return JSONObject.parseObject(value);
//            }
//        });
//
//        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonStream.assignTimestampsAndWatermarks(
//                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
//                            @Override
//                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
//                                return java.sql.Timestamp.valueOf(element.getString("create_time")).getTime();
//                            }
//                        })
//        );
//
//        // 4. 提取关键信息：(店铺ID, 退款类型, 退款金额, 处理中标记, 成功标记, 失败标记)
//        // 修复：替换 Lambda 为匿名内部类，显式声明泛型
//        SingleOutputStreamOperator<Tuple6<String, String, Double, Integer, Integer, Integer>> refundTupleStream = withWatermarkStream.map(
//                new MapFunction<JSONObject, Tuple6<String, String, Double, Integer, Integer, Integer>>() {
//                    @Override
//                    public Tuple6<String, String, Double, Integer, Integer, Integer> map(JSONObject value) throws Exception {
//                        String shopId = value.getString("shop_id");
//                        String refundType = value.getString("refund_type_code");
//                        Double amount = Double.parseDouble(value.getString("refund_amount"));
//
//                        int isProcessing = 0;
//                        int isSuccess = 0;
//                        int isFail = 0;
//
//                        if ("1502".equals(refundType)) {
//                            isSuccess = 1;
//                        } else if ("1501".equals(refundType)) {
//                            isFail = 1;
//                        } else {
//                            isProcessing = 1;
//                        }
//
//                        return new Tuple6<>(shopId, refundType, amount, isProcessing, isSuccess, isFail);
//                    }
//                }
//        );

//        // 5. 窗口聚合：统计所有指标
//        SingleOutputStreamOperator<Tuple8<String, Double, Double, Double, Integer, Integer, Long, Long>> resultStream = refundTupleStream
//                .keyBy(new KeySelector<Tuple6<String, String, Double, Integer, Integer, Integer>, String>() {
//                    @Override
//                    public String getKey(Tuple6<String, String, Double, Integer, Integer, Integer> value) throws Exception {
//                        return value.f0;
//                    }
//                })
//                .window(TumblingEventTimeWindows.of(Time.days(1)))
//                .aggregate(
//                        new AggregateFunction<
//                                Tuple6<String, String, Double, Integer, Integer, Integer>,
//                                Tuple6<String, Double, Double, Double, Integer, Integer>,
//                                Tuple6<String, Double, Double, Double, Integer, Integer>>() {
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> createAccumulator() {
//                                return new Tuple6<>("", 0.0, 0.0, 0.0, 0, 0);
//                            }
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> add(
//                                    Tuple6<String, String, Double, Integer, Integer, Integer> value,
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> accumulator) {
//
//                                double totalAmount = accumulator.f1 + value.f2;
//                                double successAmount = accumulator.f2 + (value.f4 == 1 ? value.f2 : 0);
//                                double failAmount = accumulator.f3 + (value.f5 == 1 ? value.f2 : 0);
//                                int processingCount = accumulator.f4 + value.f3;
//                                int failCount = accumulator.f5 + value.f5;
//
//                                return new Tuple6<>(value.f0, totalAmount, successAmount, failAmount, processingCount, failCount);
//                            }
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> getResult(
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> accumulator) {
//                                return accumulator;
//                            }
//
//                            @Override
//                            public Tuple6<String, Double, Double, Double, Integer, Integer> merge(
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> a,
//                                    Tuple6<String, Double, Double, Double, Integer, Integer> b) {
//                                return new Tuple6<>(
//                                        a.f0, a.f1 + b.f1, a.f2 + b.f2,
//                                        a.f3 + b.f3, a.f4 + b.f4, a.f5 + b.f5
//                                );
//                            }
//                        },
//                        new ProcessWindowFunction<
//                                Tuple6<String, Double, Double, Double, Integer, Integer>,
//                                Tuple8<String, Double, Double, Double, Integer, Integer, Long, Long>,
//                                String,
//                                TimeWindow>() {
//
//                            @Override
//                            public void process(
//                                    String shopId,
//                                    Context context,
//                                    Iterable<Tuple6<String, Double, Double, Double, Integer, Integer>> elements,
//                                    Collector<Tuple8<String, Double, Double, Double, Integer, Integer, Long, Long>> out) {
//
//                                Tuple6<String, Double, Double, Double, Integer, Integer> aggResult = elements.iterator().next();
//                                long windowStart = context.window().getStart();
//                                long windowEnd = context.window().getEnd();
//
//                                // 将 Long 时间戳转为字符串（如不做格式化，直接转字符串也可）
//                                String windowStartStr = String.valueOf(windowStart);
//                                String windowEndStr = String.valueOf(windowEnd);
//
//                                Tuple8<String, Double, Double, Double, Integer, Integer, Long, Long> result = new Tuple8<>(
//                                        aggResult.f0, aggResult.f1, aggResult.f2, aggResult.f3,
//                                        aggResult.f4, aggResult.f5, windowStartStr, windowEndStr
//                                );
//
//                                out.collect(result);
//                            }
//                        }
//                );



//        // 6. 输出结果
//        resultStream.print("各店铺退款统计：(店铺ID, 退款总金额, 退款成功金额, 退款失败金额, 处理中退款个数, 退款失败个数)");
//
//        resultStream.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_refund_shop_stats " +
//                " values(?,?,?,?,?,?,?,?,?)"));


// 3. 解析JSON并分配事件时间
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                return JSONObject.parseObject(value);
            }
        });

        SingleOutputStreamOperator<JSONObject> withWatermarkStream = jsonStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return java.sql.Timestamp.valueOf(element.getString("create_time")).getTime();
                            }
                        })
        );

// 4. 提取关键信息：(店铺ID, 退款类型, 退款金额, 处理中标记, 成功标记, 失败标记)
// 修复：替换 Lambda 为匿名内部类，显式声明泛型
        SingleOutputStreamOperator<Tuple6<String, String, Double, Integer, Integer, Integer>> refundTupleStream = withWatermarkStream.map(
                new MapFunction<JSONObject, Tuple6<String, String, Double, Integer, Integer, Integer>>() {
                    @Override
                    public Tuple6<String, String, Double, Integer, Integer, Integer> map(JSONObject value) throws Exception {
                        String shopId = value.getString("shop_id");
                        String refundType = value.getString("refund_type_code");
                        Double amount = Double.parseDouble(value.getString("refund_amount"));

                        int isProcessing = 0;
                        int isSuccess = 0;
                        int isFail = 0;

                        if ("1502".equals(refundType)) {
                            isSuccess = 1;
                        } else if ("1501".equals(refundType)) {
                            isFail = 1;
                        } else {
                            isProcessing = 1;
                        }

                        return new Tuple6<>(shopId, refundType, amount, isProcessing, isSuccess, isFail);
                    }
                }
        );

// 5. 窗口聚合：统计所有指标
// 修改输出类型为 Tuple8<..., String, String>，适配 ClickHouse 表结构
        SingleOutputStreamOperator<Tuple8<String, Double, Double, Double, Integer, Integer, String, String>> resultStream = refundTupleStream
                .keyBy(new KeySelector<Tuple6<String, String, Double, Integer, Integer, Integer>, String>() {
                    @Override
                    public String getKey(Tuple6<String, String, Double, Integer, Integer, Integer> value) throws Exception {
                        return value.f0;
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.days(1)))
                .aggregate(
                        new AggregateFunction<
                                Tuple6<String, String, Double, Integer, Integer, Integer>,
                                Tuple6<String, Double, Double, Double, Integer, Integer>,
                                Tuple6<String, Double, Double, Double, Integer, Integer>>() {

                            @Override
                            public Tuple6<String, Double, Double, Double, Integer, Integer> createAccumulator() {
                                return new Tuple6<>("", 0.0, 0.0, 0.0, 0, 0);
                            }

                            @Override
                            public Tuple6<String, Double, Double, Double, Integer, Integer> add(
                                    Tuple6<String, String, Double, Integer, Integer, Integer> value,
                                    Tuple6<String, Double, Double, Double, Integer, Integer> accumulator) {

                                double totalAmount = accumulator.f1 + value.f2;
                                double successAmount = accumulator.f2 + (value.f4 == 1 ? value.f2 : 0);
                                double failAmount = accumulator.f3 + (value.f5 == 1 ? value.f2 : 0);
                                int processingCount = accumulator.f4 + value.f3;
                                int failCount = accumulator.f5 + value.f5;

                                return new Tuple6<>(value.f0, totalAmount, successAmount, failAmount, processingCount, failCount);
                            }

                            @Override
                            public Tuple6<String, Double, Double, Double, Integer, Integer> getResult(
                                    Tuple6<String, Double, Double, Double, Integer, Integer> accumulator) {
                                return accumulator;
                            }

                            @Override
                            public Tuple6<String, Double, Double, Double, Integer, Integer> merge(
                                    Tuple6<String, Double, Double, Double, Integer, Integer> a,
                                    Tuple6<String, Double, Double, Double, Integer, Integer> b) {
                                return new Tuple6<>(
                                        a.f0, a.f1 + b.f1, a.f2 + b.f2,
                                        a.f3 + b.f3, a.f4 + b.f4, a.f5 + b.f5
                                );
                            }
                        },
                        new ProcessWindowFunction<
                                Tuple6<String, Double, Double, Double, Integer, Integer>,
                                Tuple8<String, Double, Double, Double, Integer, Integer, String, String>,
                                String,
                                TimeWindow>() {

                            @Override
                            public void process(
                                    String shopId,
                                    Context context,
                                    Iterable<Tuple6<String, Double, Double, Double, Integer, Integer>> elements,
                                    Collector<Tuple8<String, Double, Double, Double, Integer, Integer, String, String>> out) {

                                Tuple6<String, Double, Double, Double, Integer, Integer> aggResult = elements.iterator().next();
                                long windowStart = context.window().getStart();
                                long windowEnd = context.window().getEnd();

                                // 将 Long 时间戳转为字符串（直接转或格式化均可）
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                String windowStartStr = sdf.format(new Date(windowStart));
                                String windowEndStr = sdf.format(new Date(windowEnd));

                                Tuple8<String, Double, Double, Double, Integer, Integer, String, String> result = new Tuple8<>(
                                        aggResult.f0, aggResult.f1, aggResult.f2, aggResult.f3,
                                        aggResult.f4, aggResult.f5, windowStartStr, windowEndStr
                                );

                                out.collect(result);
                            }
                        }
                );

// 6. 输出结果
        resultStream.print("各店铺退款统计：(店铺ID, 退款总金额, 退款成功金额, 退款失败金额, 处理中退款个数, 退款失败个数, 窗口开始, 窗口结束)");

// 修改 Sink SQL，明确字段和占位符，与 ClickHouse 表结构完全匹配
        resultStream.addSink(MyClickHouseUtil.getSinkFunction_a(
                "INSERT INTO dws_refund_shop_stats " +
                        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
        ));


        // 9. 执行任务
        env.execute("DWS - Calculate Refund Total Amount By Shop");

    }
}
